#include "Client.h"
#include "monitor_public_def.h"
#include "ChipLog.h"
#include "rapidjson/document.h"
#include "rapidjson/prettywriter.h"
#include "rapidjson/stringbuffer.h"
using namespace  RAPIDJSON_NAMESPACE;
#include <boost/bind.hpp>


Client::Client(boost::asio::io_service& ioservice, SERVICE_CONFIG srv, STOCK_CONFIG stock, int interval, int retry)
	:m_IOservice(ioservice), m_socket(ioservice), m_deadlineTimer(ioservice)
	, m_RequestTimer(ioservice), m_strand(ioservice)/*, bExit(true)*/, m_srvcfg(srv)
	, m_stockcfg(stock), m_Interval(interval), m_Retry(retry)

{
	int bodyLength = 2 + 2 + m_stockcfg.code.length() + 10;
	PKGHeader head;
	head.attrs = 0;
	head.bootcode = 0x3c3c;
	head.seq_id = ntohl(1);
	head.type = ntohs(32947);
	head.length = ntohl(bodyLength);

	m_sendBuffer.Write((char*)&head, sizeof(PKGHeader));
	m_sendBuffer.WriteUint16(1);
	m_sendBuffer.WriteShortString(m_stockcfg.code.c_str(), m_stockcfg.code.length());
	m_sendBuffer.WriteUint32(0);
	m_sendBuffer.WriteUint16(1);
	m_sendBuffer.WriteUint32(-1);

	PKGHeader *pHead = (PKGHeader *)m_sendBuffer.GetBufferPtr();
	pHead->checksum = INetChecksum((char *)m_sendBuffer.GetBufferPtr(), sizeof(PKGHeader) + bodyLength);
}


Client::~Client()
{
}

void Client::StartWorker(std::function<void(std::string &&)> fun)
{
	funRet = fun;
	do_connect();
// 	bExit = false;

	//m_tProccess = std::thread([this]{
	//	CMemStream dataWrite;
	//	while (!bExit)
	//	{
	//		ChipData record;
	//		if (sRecordLst.Pop(&record, 3000))
	//		{
	//			dataWrite.SetSize(0);

	//			dataWrite.WriteInt32(record.iOptions);
	//			dataWrite.WriteInt16(record.sZqdmCount);

	//			dataWrite.WriteShortString(record.strZqdm);
	//			dataWrite.WriteInt8(record.i8Price_divide);
	//			dataWrite.WriteInt32(record.iDate);
	//			dataWrite.WriteInt32(record.iAvgCost);
	//			dataWrite.WriteInt32(record.iProfitCost);
	//			dataWrite.WriteInt32(record.iPressPos);
	//			dataWrite.WriteInt32(record.iSupportPos);
	//			dataWrite.WriteInt8(record.i8concenChange);
	//			dataWrite.WriteInt8(record.i8concenEvaluate);
	//			dataWrite.WriteInt8(record.i8RelativeChange);
	//			dataWrite.WriteInt8(record.i8RelativeEvaluate);
	//			dataWrite.WriteInt32(record.iTotalCapital);
	//			dataWrite.WriteInt32(record.i70High);
	//			dataWrite.WriteInt32(record.i70Low);
	//			dataWrite.WriteInt32(record.i90High);
	//			dataWrite.WriteInt32(record.i90Low);
	//			dataWrite.WriteInt16(record.sDivideNum);
	//			if (record.sDivideNum > 0)
	//			{
	//				dataWrite.Write((char *)&record.divide_price[0], sizeof(ChipData::chip_price)*record.sDivideNum);
	//			}


	//			std::string data((char *)dataWrite.GetBufferPtr(), dataWrite.GetSize());
	//			PushMQResult(0, record.idate, record.itime, data);

	//		}
	//	}
	//});
}


void Client::do_connect()
{

	boost::asio::ip::tcp::endpoint ep(boost::asio::ip::address::from_string(m_srvcfg.strIP), m_srvcfg.port);
	m_socket.async_connect(ep,
		[this](boost::system::error_code ec)
	{
		if (!ec)
		{
			do_write_buffer();
		}
		else
		{
			PushDisConnect();

			CChipLog::Instance()->WriteLog(LogLvlWarnning, "connect{%s:%d} failed, errorcode:%d, msg:%s"
				, m_srvcfg.strIP.c_str(), m_srvcfg.port, ec.value(), ec.message().c_str());
			boost::system::error_code ec;
			m_socket.close(ec);
			AsyncSleep(m_Retry);
		}
	});
}


void Client::do_request(const boost::system::error_code& error)
{
	if (!error)
	{
		if (m_socket.is_open())
		{
			m_strand.post([this]{
				boost::asio::async_write(m_socket,
					boost::asio::buffer(m_sendBuffer.GetBufferPtr(),
					m_sendBuffer.GetSize()),
					[this](boost::system::error_code ec, std::size_t){
					if (ec)
					{
						CChipLog::Instance()->WriteLog(LogLvlWarnning, "{%s:%d} do_request failed, errorcode:%d, msg:%s"
							, m_srvcfg.strIP.c_str(), m_srvcfg.port, ec.value(), ec.message().c_str());

						boost::system::error_code ec;
						m_socket.close(ec);
					}

				});
			});


		}

		m_RequestTimer.expires_from_now(boost::posix_time::seconds(m_Interval));
		m_RequestTimer.async_wait(boost::bind(&Client::do_request, this, boost::asio::placeholders::error));

	}
}

void Client::do_write_buffer()
{
	boost::asio::async_write(m_socket,
		boost::asio::buffer(m_sendBuffer.GetBufferPtr(),
		m_sendBuffer.GetSize()),
		[this](boost::system::error_code ec, std::size_t /*length*/)
	{
		if (!ec)
		{
			do_read_header();
		}
		else
		{
			PushMQResult(1, getCurrentDate(),getCurrentTime(),"");

			CChipLog::Instance()->WriteLog(LogLvlWarnning, "[%s]{%s:%d} async_write failed, errorcode:%d, msg:%s"
				, m_stockcfg.code.c_str(), m_srvcfg.strIP.c_str(), m_srvcfg.port, ec.value(), ec.message().c_str());


			boost::system::error_code ec;
			m_socket.close(ec);
			AsyncSleep(m_Retry);

		}
	});
}

void Client::do_read_header()
{
	boost::asio::async_read(m_socket, boost::asio::buffer(m_msg.GetHead(), m_msg.GetHeadLen()),
		[this](boost::system::error_code ec, std::size_t /*length*/)
	{
		if (!ec && m_msg.decode_header())
		{
			do_read_body();
		}
		else
		{
			PushDisConnect();

			CChipLog::Instance()->WriteLog(LogLvlWarnning, "[%s]{%s:%d} read_header failed, errorcode:%d, msg:%s"
				, m_stockcfg.code.c_str(), m_srvcfg.strIP.c_str(), m_srvcfg.port, ec.value(), ec.message().c_str());

			boost::system::error_code ec1;
			m_socket.close(ec1);
			AsyncSleep(m_Retry);
		}
	});
}


void Client::do_read_body()
{
	boost::asio::async_read(m_socket, boost::asio::buffer(m_msg.GetBody(), m_msg.GetBodyLen()),
		[this](boost::system::error_code ec, std::size_t /*length*/)
	{
		if (!ec )
		{
			ProccessPackage();

			boost::system::error_code ec;
			m_socket.close(ec);
			AsyncSleep(m_Interval);

		}
		else
		{
			PushDisConnect();

			CChipLog::Instance()->WriteLog(LogLvlWarnning, "[%s]{%s:%d} read_body failed, errorcode:%d, msg:%s"
				, m_stockcfg.code.c_str(), m_srvcfg.strIP.c_str(), m_srvcfg.port, ec.value(), ec.message().c_str());

			boost::system::error_code ec;
			m_socket.close(ec);
			AsyncSleep(m_Retry);
		}
	});
}


void Client::AsyncSleep(int seconds)
{
	boost::system::error_code ec;
	m_RequestTimer.cancel(ec);
	m_deadlineTimer.expires_from_now(boost::posix_time::seconds(seconds));
	m_deadlineTimer.async_wait([this](const boost::system::error_code&)
	{
		do_connect();
	});
}

void Client::ProccessPackage()
{

	if (!m_msg.decode_body())
	{
		CChipLog::Instance()->WriteLog(LogLvlWarnning, "[%s]{%s:%d} decode body failed: checksum failed"
			, m_stockcfg.code.c_str(), m_srvcfg.strIP.c_str(), m_srvcfg.port);
		return;
	}
	PKGHeader* pHead = (PKGHeader*)m_msg.GetHead();
	if (pHead->type != 32947 || pHead->error_code != 0)
	{
		CChipLog::Instance()->WriteLog(LogLvlWarnning, "[%s]{%s:%d} decode body failed,type=%d,error_code=%d "
			, m_stockcfg.code.c_str(), m_srvcfg.strIP.c_str(), m_srvcfg.port, pHead->type, pHead->error_code);
		return;
	}

	CHisStream dataStream(m_msg.GetBody(), m_msg.GetBodyLen());
	ChipData record;
	record.idate = getCurrentDate();
	record.itime = getCurrentTime();
	dataStream.ReadInt32(&record.iSubPackage);
	dataStream.ReadInt32(&record.iOptions);
	dataStream.ReadInt16(&record.sZqdmCount);

	if (record.sZqdmCount > 0)
	{
		dataStream.ReadShortString(record.strZqdm, 22);

		dataStream.ReadInt8(&record.i8Price_divide);

		dataStream.ReadInt32(&record.iDate);

		if (((record.iOptions >> 0) & 0x1) == 0x1)
			dataStream.ReadInt32(&record.iAvgCost);

		if (((record.iOptions >> 1) & 0x1) == 0x1)
			dataStream.ReadInt32(&record.iProfitCost);

		if (((record.iOptions >> 2) & 0x1) == 0x1)
		{
			dataStream.ReadInt32(&record.iPressPos);
			dataStream.ReadInt32(&record.iSupportPos);
		}

		if (((record.iOptions >> 3) & 0x1) == 0x1)
		{
			dataStream.ReadInt8(&record.i8concenChange);
			dataStream.ReadInt8(&record.i8concenEvaluate);
		}

		if (((record.iOptions >> 4) & 0x1) == 0x1)
		{
			dataStream.ReadInt8(&record.i8RelativeChange);
			dataStream.ReadInt8(&record.i8RelativeEvaluate);
		}

		if (((record.iOptions >> 5) & 0x1) == 0x1)
			dataStream.ReadInt32(&record.iTotalCapital);

		if (((record.iOptions >> 6) & 0x1) == 0x1)
		{
			dataStream.ReadInt32(&record.i70High);
			dataStream.ReadInt32(&record.i70Low);
		}
		if (((record.iOptions >> 7) & 0x1) == 0x1)
		{
			dataStream.ReadInt32(&record.i90High);
			dataStream.ReadInt32(&record.i90Low);
		}

		if (((record.iOptions >> 8) & 0x1) == 0x1)
		{
			dataStream.ReadInt16(&record.sDivideNum);

			if (record.sDivideNum > 0)
			{
				record.divide_price.resize(record.sDivideNum);
			}

			for (int i = 0; i < record.sDivideNum; i++)
			{

				dataStream.ReadInt32(&record.divide_price[i].iPrice);
				dataStream.ReadInt32(&record.divide_price[i].iBuy);
				dataStream.ReadInt32(&record.divide_price[i].iSell);
			}


		}

	}


	dataWrite.SetSize(0);

	dataWrite.WriteInt32(record.iOptions);
	dataWrite.WriteInt16(record.sZqdmCount);

	dataWrite.WriteShortString(record.strZqdm);
	dataWrite.WriteInt8(record.i8Price_divide);
	dataWrite.WriteInt32(record.iDate);
	dataWrite.WriteInt32(record.iAvgCost);
	dataWrite.WriteInt32(record.iProfitCost);
	dataWrite.WriteInt32(record.iPressPos);
	dataWrite.WriteInt32(record.iSupportPos);
	dataWrite.WriteInt8(record.i8concenChange);
	dataWrite.WriteInt8(record.i8concenEvaluate);
	dataWrite.WriteInt8(record.i8RelativeChange);
	dataWrite.WriteInt8(record.i8RelativeEvaluate);
	dataWrite.WriteInt32(record.iTotalCapital);
	dataWrite.WriteInt32(record.i70High);
	dataWrite.WriteInt32(record.i70Low);
	dataWrite.WriteInt32(record.i90High);
	dataWrite.WriteInt32(record.i90Low);
	dataWrite.WriteInt16(record.sDivideNum);
	if (record.sDivideNum > 0)
	{
		dataWrite.Write((char *)&record.divide_price[0], sizeof(ChipData::chip_price)*record.sDivideNum);
	}


	std::string data((char *)dataWrite.GetBufferPtr(), dataWrite.GetSize());
	PushMQResult(0, record.idate, record.itime, data);

		


// 	sRecordLst.PushBack(std::forward<ChipData>(record));


}

void Client::PushDisConnect()
{
	PushMQResult(1, getCurrentDate(), getCurrentTime2(), "");

}

void Client::PushConnect()
{
	PushMQResult(2, getCurrentDate(), getCurrentTime2(), "");
	
}


void Client::PushMQResult(int status, int date, int time, const std::string &data)
{
	Document root;
	RAPIDJSON_NAMESPACE::Document::AllocatorType &allocator = root.GetAllocator();
	root.SetObject();

	root.AddMember(("msg"), 0x03, allocator);
	root.AddMember("sign", StringRef(m_srvcfg.strSign.c_str()), allocator);
	root.AddMember("ip", StringRef(m_srvcfg.strIP.c_str()), allocator);
	root.AddMember("port", m_srvcfg.port, allocator);
	root.AddMember("date", date, allocator);
	root.AddMember("time", time, allocator);
	root.AddMember("code", StringRef(m_stockcfg.code.c_str()), allocator);
	root.AddMember("status", status, allocator);

	RAPIDJSON_NAMESPACE::Value text(RAPIDJSON_NAMESPACE::kObjectType);
	text.AddMember("data", StringRef(data.c_str(), data.size()), allocator);

	root.AddMember("text", text, allocator);

	if (funRet)
	{
		StringBuffer sbBuf;
		Writer<StringBuffer> jWriter(sbBuf);
		root.Accept(jWriter);
		funRet(sbBuf.GetString());
	}
}
